package com.miao.im.service.message;

import cn.hutool.core.collection.CollectionUtil;
import com.alibaba.fastjson.JSON;
import com.miao.im.common.CacheKey;
import com.miao.im.repository.IMessageRepository;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.Range;
import org.springframework.data.redis.connection.RedisZSetCommands;
import org.springframework.data.redis.core.ReactiveStringRedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;
import reactor.core.publisher.Mono;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * @Auther: miaoguoxin
 * @Date: 2021/3/18 14:38
 * @Description:管理离线消息
 */
@Component
@Slf4j
public class OfflineMessageManager implements IMessageManager {
    @Autowired
    private StringRedisTemplate redisTemplate;
    @Autowired
    private ReactiveStringRedisTemplate reactiveStringRedisTemplate;
    @Autowired
    private IMessageRepository messageRepository;

    @Override
    public Mono<Void> save(ChatMessage chatMessage) {
        return messageRepository.reactiveAddOfflineMsg(chatMessage)
                .then(messageRepository.reactiveIncOfflineCounter(chatMessage));
    }

    @Override
    public Mono<List<ChatMessage>> getPersonalPage(Long userId, Long targetId, Integer targetType, long timestamp, Integer pageSize) {
        if (timestamp == 0) {
            timestamp = System.currentTimeMillis();
        }
        String key = getKey(userId, targetId, targetType);
        Range<Double> closed = Range.closed(0d, (double) timestamp);
        RedisZSetCommands.Limit limit = RedisZSetCommands.Limit.limit().offset(0).count(pageSize);
        return reactiveStringRedisTemplate.opsForZSet()
                .reverseRangeByScore(key, closed, limit)
                .collectList()
                .flatMap(jsonList -> {
                    if (CollectionUtil.isEmpty(jsonList)) {
                        return Mono.just(new ArrayList<>());
                    }
                    return Mono.just(jsonList.stream()
                            .map(s -> JSON.parseObject(s, ChatMessage.class))
                            .collect(Collectors.toList()));
                });
    }

    @Override
    public Map<String, OfflineMessageCounter> getOfflineMsgCounter(Long userId) {
        return messageRepository.getMsgCounter(userId);
    }

    @Override
    public void delOfflineCounter(Long userId) {
        String key = CacheKey.OFFLINE_MSG_COUNTER_PREFIX + userId;
        redisTemplate.delete(key);
    }

    @Override
    public Mono<Void> deletePersonalRange(Long userId, Long targetId, Integer targetType, long startTimestamp, long endTimestamp) {
        String key = getKey(userId, targetId, targetType);
        Range<Double> range = Range.closed((double) startTimestamp, (double) endTimestamp);
        return reactiveStringRedisTemplate.opsForZSet().removeRangeByScore(key, range).then();
    }

    private static String getKey(Long userId, Long targetId, Integer targetType) {
        return CacheKey.OFFLINE_MSG_PREFIX + targetId +
                CacheKey.SEMICOLON_DELIMITER + targetType +
                CacheKey.SEMICOLON_DELIMITER + userId;
    }

}
